package com.ctsi.ftp.server;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import com.ctsi.ftp.thread.AccountAlarmTask;
import com.ctsi.ftp.util.*;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.Path;

import org.apache.hadoop.fs.PathFilter;


public class UserLoggerService {
    private static Log log = LogFactory.getLog(UserLoggerService.class);

    private static boolean stop = false;

    public static String FILE_SEP = "/";

    public static int FILENAME_MAX_LENGTH=Integer.parseInt(ConfigProperties.getProperty("fileNameMaxLength"));
//	private static Object lockObj = new Object();

    private static long executiveTime = Long.parseLong(ConfigProperties.getProperty("executiveTime"));

    private static int maxUploadFileThreadNum = Integer.parseInt(ConfigProperties.getProperty("maxUploadFileThreadNum"));

    private ExecutorService exec = Executors.newFixedThreadPool(maxUploadFileThreadNum);

    private String ftpUrl;

    private int ftpPort;

    private String ftpUserName;

    private String ftpPassword;

    private boolean isGZip;

    public UserLoggerService(String ftpUrl, int ftpPort, String ftpUserName,
                             String ftpPassword) {
        super();
        this.ftpUrl = ftpUrl;
        this.ftpPort = ftpPort;
        this.ftpUserName = ftpUserName;
        this.ftpPassword = ftpPassword;
    }

    public UserLoggerService(String ftpUrl, int ftpPort, String ftpUserName,
                             String ftpPassword, boolean isGZip) {
        super();
        this.ftpUrl = ftpUrl;
        this.ftpPort = ftpPort;
        this.ftpUserName = ftpUserName;
        this.ftpPassword = ftpPassword;
        this.isGZip = isGZip;
    }


    public void send(String hdfsRootPath, String remoteRootPath, String filePostfix, int threadNum) {
        while (!stop) {
            // 实际处理过程

            log.info("The userLoggerService  " + threadNum + "  is send...");
            try {
                Thread.currentThread();
                Thread.sleep(executiveTime * 1000);
            } catch (InterruptedException e1) {
                e1.printStackTrace();
            }
            try {
                //获取当前时间
                long time = System.currentTimeMillis();
                //获取遍历的最长天数
                long maxTime = Long.parseLong(ConfigProperties.getProperty("maxTime"));
                // 用户日志文件上传服务每条线程一次上传的文件数
                int maxPerUploadFileNum = Integer.parseInt(ConfigProperties.getProperty("maxPerUploadFileNum"));


                for (long sum = maxTime; sum >= 0; sum--) {
                    Date currentTime = new Date(time - sum * 24 * 60 * 60 * 1000l);
                    SimpleDateFormat sdf = new SimpleDateFormat("yyyy" + "MM" + "dd");
                    String hdfsPath = hdfsRootPath + FILE_SEP + sdf.format(currentTime);
//                    String successHdfsPath = hdfsPath + FILE_SEP + "bak";
                    String successHdfsPath = hdfsRootPath + FILE_SEP + "_bak";
                    log.debug("The hdfs success path :"+successHdfsPath);
                    int filesLen = 0;
                    if (!HdfsUtil.existsHdfsPath(new Path(hdfsPath))) {
                        log.warn("The directory [" + hdfsPath + "] not exist!");
                        continue;
                    }
                    /**从HDFS中获取文件目录列表**/
                    //遍历目录（按小时划分）

                    PathFilter dirFilter = new DirFilter();
                    PathFilter fileFilter = new FileFilter();


                    FileStatus[] hourDirPath = HdfsUtil.listHdfsDir(hdfsPath, dirFilter);
                    ArrayList<Path> files = new ArrayList<Path>();
                    for (int i = 0; i < hourDirPath.length; i++) {
                          //log.info("The hourDir:" + hourDirPath[i].getPath().toString());
//                        if (HdfsUtil.existsHdfsPath(hourDirPath[i].getPath()) &&
//                                HdfsUtil.isHdfsDirectory(hourDirPath[i].getPath())){
                        FileStatus[] interDirPath = HdfsUtil.listHdfsDir(hourDirPath[i].getPath(), dirFilter);
                        //遍历目录（按时间间隔划分）
                        for (int j = 0; j < interDirPath.length; j++) {
                            //*****************************
//                                if (HdfsUtil.existsHdfsPath(interDirPath[i].getPath()) &&
//                                        HdfsUtil.isHdfsDirectory(interDirPath[i].getPath())){
                             log.debug("The intervalDir:" + interDirPath[j].getPath().toString());
                            // 获取文件

                            FileStatus[] fileStatus = HdfsUtil.listHdfsDir(interDirPath[j].getPath(), fileFilter);
                            for (int k = 0; k < fileStatus.length; k++) {
                                Path file = fileStatus[k].getPath();

                                if (HdfsUtil.isHdfsFile(file)) {
                                    //判断是否为success文件，如果是则直接删除
                                    if (file.getName().endsWith("_SUCCESS")) {
                                        HdfsUtil.deleteDir(file);
                                        continue;
                                    }
                                    Path dirFile = new Path(file.getParent() + "/" + System.currentTimeMillis() + file.getName());
                                   //SXS:avoid too long filename
                                    if(file.getName().length()<FILENAME_MAX_LENGTH) {
                                        HdfsUtil.renameHdfsFile(file, dirFile);
                                        files.add(dirFile);
                                        //*****************************
                                        log.info("Add file the first time:" + dirFile.getName().toString());
                                    }else{
                                        files.add(file);
                                        //*****************************
                                       log.info("Add file again:" + file.getName().toString());
                                    }
                                }
                            }
//                                }
                        }
//                        }
                    }

                    filesLen = files.size();  //文件个数
//						***********debug info********
                    log.info("The total file number:"+filesLen);
//						记录待上传文件数，值从0到MAX上传文件数
                    int uploadFilesSum = 0;
//						记录待上传的文件，数组大小为MAX上传文件数
                    Path[] uploadfiles = new Path[maxPerUploadFileNum];
//						已经上传的总文件数目
                    int uploadedFilesSum = 0;
//						以启动线程数
                    int taskNum = 0;
//						计数器,用于实现:所有子线程都结束时,主线程才能继续往下执行
                    CountDownLatch countDownLatch;
//						计数器的值的算法:取同时运行的最大线程数和[(总文件数-以上传文件数)/每次任务处理的最大文件数,能整除,则取商值,不能整除,则取商值加1]之间的最小值
                    int currentThreadNum = Math.min(maxPerUploadFileNum,
                            (filesLen - uploadedFilesSum) % maxPerUploadFileNum == 0 ? (filesLen - uploadedFilesSum) / maxPerUploadFileNum
                                    : ((filesLen - uploadedFilesSum) / maxPerUploadFileNum + 1));
                    countDownLatch = new CountDownLatch(currentThreadNum);
//						是否已经传完文件夹下的所有文件
                    boolean isLastFile = false;
                    while (!files.isEmpty()) {
                        /**
                         * 三.开始遍历当前目录下,取到的所有文件
                         */
                        Path file = files.get(0);
                        files.remove(0);
//							将文件加入到待上传的文件数组中
                        uploadfiles[uploadFilesSum] = file;
//							待上传文件数加1
                        uploadFilesSum++;
//							已经上传的总文件数目加1
                        uploadedFilesSum++;
                        if (uploadedFilesSum == filesLen) {
                            isLastFile = true;
                        }
                        if (uploadFilesSum == maxPerUploadFileNum || isLastFile) {
//								两种情况起线程发文件：一，当前待上传文件数达到MAX上传文件数最大；二，已经上传的文件数目等于当前目录下，时间戳以后的文件数目
                            String taskId = "(" + taskNum + ")-{" + hdfsPath + "}";
                            AccountAlarmTask accountAlarmTask = new AccountAlarmTask(taskId, uploadfiles, ftpUrl, ftpPort, ftpUserName, ftpPassword
                                    , remoteRootPath, successHdfsPath, filePostfix, isGZip, countDownLatch);

//								启动线程
                            exec.execute(accountAlarmTask);
//								以启动线程数加1
                            taskNum++;
//								待上传文件数清0
                            uploadFilesSum = 0;
//								待上传文件数组清空
                            uploadfiles = new Path[maxPerUploadFileNum];
                            if (taskNum == maxUploadFileThreadNum) {
                                /**
                                 * 当以启动线程数等于同时运行的最大线程数,通过计数器,让主线程等待所有子线程的结束
                                 */
                                log.debug("begin " + taskNum);
                                try {
                                    countDownLatch.await();
                                } catch (InterruptedException e) {
                                    e.printStackTrace();
                                } finally {
                                    log.debug("=========== " + countDownLatch.getCount());
                                }
                                log.debug("end " + taskNum);
//									重置计数器
                                currentThreadNum = Math
                                        .min(
                                                maxPerUploadFileNum,
                                                (filesLen - uploadedFilesSum) % maxPerUploadFileNum == 0
                                                        ? (filesLen - uploadedFilesSum) / maxPerUploadFileNum
                                                        : ((filesLen - uploadedFilesSum) / maxPerUploadFileNum + 1));
                                countDownLatch = new CountDownLatch(currentThreadNum);
                                log.debug("all files num :" + filesLen + "uploaded files : " + uploadedFilesSum + ";reset next loop ,start thead num : " + currentThreadNum);
//									重置以启动线程数
                                taskNum = 0;
                            } else {
                                if (isLastFile) {
                                    /**
                                     * 当处理的是最后一个文件,通过计数器,让主线程等待所有子线程的结束
                                     */
                                    files.clear();
                                    log.debug("begin " + taskNum);
                                    try {
                                        countDownLatch.await();
                                    } catch (InterruptedException e) {
                                        e.printStackTrace();
                                    } finally {
                                        log.debug("=========== " + countDownLatch.getCount());
                                    }
                                    log.debug("end " + taskNum);
                                }
                                continue;
                            }
                        }
                    }

                }
            } catch (Exception e) {
                e.printStackTrace();
            }
//			}
        }
    }


    public void stop() {
        stop = true;
//		synchronized (lockObj) {
//			lockObj.notifyAll();
//		}
        log.debug("UserLoggerService Service Stoped.");
        stop = false;
    }
}
